package br.pucrs.lognplay.hadoop.test;

import java.io.IOException;

import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.Path;
import org.apache.log4j.Logger;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.search.SearchHit;

import br.pucrs.lognplay.hadoop.ElasticSearchClient;
import br.pucrs.lognplay.hadoop.HDFSTools;

public class Archiver {

	private static final String ES_HOST = "ec2-177-71-144-69.sa-east-1.compute.amazonaws.com";

	private static final Logger logger = Logger.getLogger(Archiver.class);

	public static void main(String[] args) {
		archive("lognplay", "logmsg", "client-ADP", "logSource-Odyssey1");
	}

	private static void archive(String view, String type, String client, String logSource) {
		ElasticSearchClient esClient = new ElasticSearchClient();
		SearchResponse scrollResp = esClient.getEsClient().prepareSearch(view).setSearchType(SearchType.SCAN).setTypes(type).setScroll(new TimeValue(60000)).setSize(100).execute().actionGet();

		try {
			HDFSTools.mkdir("/usr/lognplay/" + client + "/");
			FSDataOutputStream fileOut = HDFSTools.getFileSystem().create(new Path("/usr/lognplay/" + client + "/" + logSource + ".log"));
			while (true) {
				scrollResp = esClient.getEsClient().prepareSearchScroll(scrollResp.getScrollId()).setScroll(new TimeValue(600000)).execute().actionGet();
				for (SearchHit hit : scrollResp.getHits()) {
					logger.trace(hit.getSource().toString());
					fileOut.write(hit.getSource().toString().getBytes());
				}
				if (scrollResp.hits().hits().length == 0) {
					break;
				}
			}
			fileOut.close();
			HDFSTools.getFileSystem().close();
		} catch (IOException e) {
			e.printStackTrace();
		}
		logger.trace("Closing connection");
		esClient.getEsClient().close();
		logger.trace("Connection is closed");
	}

}
